package org.example.app.dws;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.app.bean.SearchKeyword;
import org.example.app.bean.SearchRankResult;
import org.example.app.bean.SearchSalesAccumulator;
import org.example.app.function.SearchRankProcessFunction;
import org.example.utils.MyClickHouseUtil;
import org.example.utils.MyKafkaUtil;
import org.example.utils.SystemConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 搜索排行榜处理器 - 符合实时数仓设计规范
 * 实现标准化的DWS层数据处理流程
 */
public class DwsMarketSearchRank {
    
    private static final Logger logger = LoggerFactory.getLogger(DwsMarketSearchRank.class);
    private static final String DATA_SOURCE = "kafka_dwd_market_search";
    private static final Integer VERSION = 1;
    
    public static void main(String[] args) throws Exception {
        // 初始化系统配置，避免Hadoop路径问题
        SystemConfigUtil.initializeSystemConfig();
        
        logger.info("启动DwsMarketSearchRank...");
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "dwd_market_search";
        String groupId = "dws_market_search_rank_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 转换为SearchKeyword对象并添加水位线
        SingleOutputStreamOperator<SearchKeyword> keywordDS = kafkaDS.map(new MapFunction<String, SearchKeyword>() {
            @Override
            public SearchKeyword map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                SearchKeyword keyword = new SearchKeyword();
                keyword.setKeyword(jsonObject.getString("keyword"));
                keyword.setSearchCount(jsonObject.getLong("search_volume"));
                keyword.setClickRate(jsonObject.getDouble("click_rate"));
                keyword.setConversionRate(jsonObject.getDouble("conversion_rate"));
                keyword.setDemandSupplyRatio(jsonObject.getDouble("supply_demand_ratio"));
                keyword.setCategoryId(jsonObject.getString("category_id"));
                keyword.setCategoryName(jsonObject.getString("category_name"));
                return keyword;
            }
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<SearchKeyword>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<SearchKeyword>() {
                    @Override
                    public long extractTimestamp(SearchKeyword element, long recordTimestamp) {
                        return System.currentTimeMillis();
                    }
                })
        );
        
        // 按关键词分组
        KeyedStream<SearchKeyword, String> keyedStream = keywordDS.keyBy(new KeySelector<SearchKeyword, String>() {
            @Override
            public String getKey(SearchKeyword value) throws Exception {
                return value.getKeyword();
            }
        });
        
        // 1. 计算搜索热度排行榜
        SingleOutputStreamOperator<SearchRankResult> searchHotRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(1)))
            .aggregate(new SearchHotAggregateFunction(), new SearchHotWindowFunction())
            .keyBy(search -> "search_hot")
            .process(new SearchRankProcessFunction("search_hot", 100));
        
        // 2. 计算搜索交易排行榜
        SingleOutputStreamOperator<SearchRankResult> searchTransactionRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(1)))
            .aggregate(new SearchTransactionAggregateFunction(), new SearchTransactionWindowFunction())
            .keyBy(search -> "search_transaction")
            .process(new SearchRankProcessFunction("search_transaction", 100));
        
        // 3. 计算搜索点击排行榜
        SingleOutputStreamOperator<SearchRankResult> searchClickRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(1)))
            .aggregate(new SearchClickAggregateFunction(), new SearchClickWindowFunction())
            .keyBy(search -> "search_click")
            .process(new SearchRankProcessFunction("search_click", 100));
        
        // 4. 计算蓝海词排行榜
        SingleOutputStreamOperator<SearchRankResult> blueOceanRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new BlueOceanAggregateFunction(), new BlueOceanWindowFunction())
            .keyBy(search -> "blue_ocean")
            .process(new SearchRankProcessFunction("blue_ocean", 100));
        
        // 将结果写入ClickHouse - 使用符合实时数仓规范的SQL
        searchHotRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_search_hot_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        searchTransactionRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_search_transaction_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        searchClickRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_search_click_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        blueOceanRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_search_blue_ocean_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        // 执行任务
        env.execute("Search Ranking Job");
    }
    
    // 搜索热度聚合函数
    public static class SearchHotAggregateFunction implements AggregateFunction<SearchKeyword, SearchSalesAccumulator, SearchSalesAccumulator> {
        @Override
        public SearchSalesAccumulator createAccumulator() {
            return new SearchSalesAccumulator();
        }
        
        @Override
        public SearchSalesAccumulator add(SearchKeyword value, SearchSalesAccumulator accumulator) {
            accumulator.keyword = value.getKeyword();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSearchCount += value.getSearchCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator getResult(SearchSalesAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator merge(SearchSalesAccumulator a, SearchSalesAccumulator b) {
            a.totalSearchCount += b.totalSearchCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 搜索热度窗口函数
    public static class SearchHotWindowFunction implements WindowFunction<SearchSalesAccumulator, SearchRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<SearchSalesAccumulator> input, Collector<SearchRankResult> out) throws Exception {
            SearchSalesAccumulator accumulator = input.iterator().next();
            
            SearchRankResult result = new SearchRankResult();
            // 设置业务字段
            result.setKeyword(accumulator.keyword);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置度量字段
            result.setSearchCount(accumulator.totalSearchCount);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 搜索交易聚合函数
    public static class SearchTransactionAggregateFunction implements AggregateFunction<SearchKeyword, SearchSalesAccumulator, SearchSalesAccumulator> {
        @Override
        public SearchSalesAccumulator createAccumulator() {
            return new SearchSalesAccumulator();
        }
        
        @Override
        public SearchSalesAccumulator add(SearchKeyword value, SearchSalesAccumulator accumulator) {
            accumulator.keyword = value.getKeyword();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSearchCount += value.getSearchCount();
            // 模拟交易数据
            accumulator.totalSalesAmount += value.getSearchCount() * value.getConversionRate() * 100; // 假设平均单价100
            accumulator.totalOrderCount += (long)(value.getSearchCount() * value.getConversionRate());
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator getResult(SearchSalesAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator merge(SearchSalesAccumulator a, SearchSalesAccumulator b) {
            a.totalSearchCount += b.totalSearchCount;
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalOrderCount += b.totalOrderCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 搜索交易窗口函数
    public static class SearchTransactionWindowFunction implements WindowFunction<SearchSalesAccumulator, SearchRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<SearchSalesAccumulator> input, Collector<SearchRankResult> out) throws Exception {
            SearchSalesAccumulator accumulator = input.iterator().next();
            
            SearchRankResult result = new SearchRankResult();
            // 设置业务字段
            result.setKeyword(accumulator.keyword);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置度量字段
            result.setSearchCount(accumulator.totalSearchCount);
            result.setOrderCount(accumulator.totalOrderCount);
            
            // 计算转化率
            if (accumulator.totalSearchCount > 0) {
                result.setConversionRate(accumulator.totalOrderCount.doubleValue() / accumulator.totalSearchCount * 100);
            } else {
                result.setConversionRate(0.0);
            }
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 搜索点击聚合函数
    public static class SearchClickAggregateFunction implements AggregateFunction<SearchKeyword, SearchSalesAccumulator, SearchSalesAccumulator> {
        @Override
        public SearchSalesAccumulator createAccumulator() {
            return new SearchSalesAccumulator();
        }
        
        @Override
        public SearchSalesAccumulator add(SearchKeyword value, SearchSalesAccumulator accumulator) {
            accumulator.keyword = value.getKeyword();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSearchCount += value.getSearchCount();
            accumulator.totalClickCount += (long)(value.getSearchCount() * value.getClickRate());
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator getResult(SearchSalesAccumulator accumulator) {
            accumulator.calculateAvgClickRate();
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator merge(SearchSalesAccumulator a, SearchSalesAccumulator b) {
            a.totalSearchCount += b.totalSearchCount;
            a.totalClickCount += b.totalClickCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 搜索点击窗口函数
    public static class SearchClickWindowFunction implements WindowFunction<SearchSalesAccumulator, SearchRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<SearchSalesAccumulator> input, Collector<SearchRankResult> out) throws Exception {
            SearchSalesAccumulator accumulator = input.iterator().next();
            
            SearchRankResult result = new SearchRankResult();
            // 设置业务字段
            result.setKeyword(accumulator.keyword);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置度量字段
            result.setSearchCount(accumulator.totalSearchCount);
            result.setClickCount(accumulator.totalClickCount);
            result.setClickRate(accumulator.avgClickRate);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 蓝海词聚合函数
    public static class BlueOceanAggregateFunction implements AggregateFunction<SearchKeyword, SearchSalesAccumulator, SearchSalesAccumulator> {
        @Override
        public SearchSalesAccumulator createAccumulator() {
            return new SearchSalesAccumulator();
        }
        
        @Override
        public SearchSalesAccumulator add(SearchKeyword value, SearchSalesAccumulator accumulator) {
            accumulator.keyword = value.getKeyword();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSearchCount += value.getSearchCount();
            // 蓝海词分析：需求供给比
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator getResult(SearchSalesAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public SearchSalesAccumulator merge(SearchSalesAccumulator a, SearchSalesAccumulator b) {
            a.totalSearchCount += b.totalSearchCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 蓝海词窗口函数
    public static class BlueOceanWindowFunction implements WindowFunction<SearchSalesAccumulator, SearchRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<SearchSalesAccumulator> input, Collector<SearchRankResult> out) throws Exception {
            SearchSalesAccumulator accumulator = input.iterator().next();
            
            SearchRankResult result = new SearchRankResult();
            // 设置业务字段
            result.setKeyword(accumulator.keyword);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置度量字段
            result.setSearchCount(accumulator.totalSearchCount);
            
            // 蓝海词特征：搜索量适中但竞争较小
            // 这里简化为搜索量在500-5000之间的词
            if (accumulator.totalSearchCount >= 500 && accumulator.totalSearchCount <= 5000) {
                result.setDemandSupplyRatio(accumulator.totalSearchCount.doubleValue()); // 需求供给比作为蓝海分数
            } else {
                result.setDemandSupplyRatio(0.0);
            }
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
}